<?php
/**
 * workerman
 * chenweizhe
 */

namespace Xtsb\Cims\Push;

use think\facade\Log;
use think\worker\Server;
use Workerman\Connection\TcpConnection;
use Workerman\Lib\Timer;
use Workerman\Worker as Workerman;

class WsPush extends Server
{
  protected $socket = 'websocket://127.0.0.1:2346';
  protected $name = 'xtsb-cims';//当前Worker实例的名称
  protected $option = [
    'name' => 'XtsbWebsocketWorker',
    'count' => 12,//进程数
    'connections' => [],//存储了当前进程的所有的客户端连接对象 array(id=>connection, id=>connection, ...)
  ];

  protected $heartbeatTime = 60;
  protected $uidConnectons = [];//存储了当前进程的所有的客户端连接对象

  protected $worker = [];//worker

//  protected function init()
//  {
//    parent::init(); // TODO: Change the autogenerated stub
//
//    $this->socket = 'websocket://' . request()->ip() . ':2345';
//  }

  /**
   * 进程启动回调
   * php think worker:server start
   * @param $worker
   */
  public function onWorkerStart($worker)
  {
//    dump('WorkerMan start');
    $this->worker = $worker;

    $this->heartbeat($worker);//心跳

    $this->innerTextWorker();//消费消息
  }

  /**
   * 当客户端与Workerman建立连接时(TCP三次握手完成后)触发的回调函数
   * @param $connection
   */
  public function onConnect($connection)
  {
    //$connection->getRemoteIp()获得对方ip,没有其他可以鉴别客户端的数据或者信息
//    Log::channel('ws')->info('websocket 连接成功：{data}', ['data' => 'IP:【' . $connection->getRemoteIp() . '】']);

    $connection->send(json_encode(['code' => 0, 'msg' => '客户端连接成功']));
  }

  /**
   * 当客户端通过连接发来数据时(Workerman收到数据时)触发的回调函数
   * @param TcpConnection $connection 连接对象，即TcpConnection实例，用于操作客户端连接，如发送数据，关闭连接等
   * @param string $data 客户端连接上发来的数据
   * @return void
   */
  public function onMessage(TcpConnection $connection, string $data)
  {
//    dump('收到的客户端数据：' . $data);
    Log::channel('ws')->info('websocket 收到客户端数据：{data}', ['data' => $data]);

    $worker = $this->worker;
    // 判断当前客户端是否已经验证,既是否设置了uid
    if (!isset($connection->uid)) {
      $dat = json_decode($data, true);
      //保存UID
      if (!isset($dat['uid'])) {
        $connection->send(json_encode(['code' => 1, 'msg' => '客户端连接失败，UID未上传']));
        return;
      }
      // 没验证的话把第一个包当做uid（这里为了方便演示，没做真正的验证）
      $connection->uid = $dat['uid'];
      $connection->lastMessageTime = time();//最新收到信息时间
      /* 保存uid到connection的映射，这样可以方便的通过uid查找connection，
       * 实现针对特定uid推送数据
       */
      $worker->uidConnections[$connection->uid] = $connection;
//      dump('-----------onMessage start-----------', $connection, $data, '-----------onMessage end-----------');
    } else {
      //... 更新心跳时间
      $connection->lastMessageTime = time();//最新收到信息时间
    }
  }


  /**
   * 当连接断开时触发的回调函数
   * @param $connection
   */
  public function onClose($connection)
  {
    $worker = $this->worker;
    if (isset($connection->uid)) {
      // 连接断开时删除映射
      unset($worker->uidConnections[$connection->uid]);
    }
  }

  /**
   * 当客户端的连接上发生错误时触发
   * @param $connection
   * @param $code
   * @param $msg
   */
  public function onError($connection, $code, $msg)
  {
    echo "error $code $msg\n";
  }

  //向uid发送消息
  protected function sendMessageByUid($uid, $message)
  {
    $worker = $this->worker;
    if (isset($worker->uidConnections[$uid])) {
      $connection = $worker->uidConnections[$uid];
      $connection->send($message);

      Log::channel('ws')->info('websocket 推送成功：{data}', ['data' => 'UID:【' . $uid . '】' . $message]);
      return true;
    } else {
      Log::channel('ws')->info('websocket 推送失败：{data}', ['data' => 'UID:【' . $uid . '】离线']);
      return false;
    }
  }

  // 向所有验证的用户推送数据
  function broadcast($message)
  {
    $worker = $this->worker;
    foreach ($worker->uidConnections as $connection) {
      $connection->send($message);
    }
  }

  /**
   * 心跳监听
   * 建议客户端发送心跳间隔小于60秒，比如55秒
   * @param $worker
   * @return void
   */
  protected function heartbeat($worker)
  {
    Timer::add(55, function () use ($worker) {
      $time_now = time();
      foreach ($worker->connections as $connection) {
        // 有可能该connection还没收到过消息，则lastMessageTime设置为当前时间
        if (empty($connection->lastMessageTime)) {
          $connection->lastMessageTime = $time_now;
          continue;
        }
        // 上次通讯时间间隔大于心跳间隔，则认为客户端已经下线，关闭连接
        if ($time_now - $connection->lastMessageTime > $this->heartbeatTime) {
          $connection->close();
        }
      }
    });
  }

  /**
   * 示例 php后端及时推送消息给客户端
   * 原理：
   * 1、建立一个websocket Worker，用来维持客户端长连接
   * 2、websocket Worker内部建立一个text Worker
   * 3、websocket Worker 与 text Worker是同一个进程，可以方便的共享客户端连接
   * 4、某个独立的php后台系统通过text协议与text Worker通讯
   * 5、text Worker操作websocket连接完成数据推送
   * @return void
   */
  protected function innerTextWorker()
  {
    // 开启一个内部端口，方便内部系统推送数据，Text协议格式 文本+换行符
    $innerTextWorker = new Workerman('text://127.0.0.1:5678');
    $innerTextWorker->onMessage = function (TcpConnection $connection, $buffer) {
      // $data数组格式，里面有uid，表示向那个uid的页面推送数据
      $data = json_decode($buffer, true);
      $uid = $data['uid'];
      // 通过workerman，向uid的页面推送数据
      $ret = $this->sendMessageByUid($uid, $buffer);
      // 返回推送结果
      $connection->send($ret ? 'ok' : 'fail');
    };
    // ## 执行监听 ##
    $innerTextWorker->listen();
  }

  //推送信息
  /**
   * @param $content array 消息内容
   * @param $noticeType string 消息类型
   * @return void
   */
  //$data=['tag' => 'notify', 'uid' => 294, 'content' => [], 'msg' => 'success']
  public static function push($content, $uid, $noticeType = 'notify')
  {
    $errCode = 1;
    $errMsg = '网络错误';
    // 建立socket连接到内部推送端口
    $client = stream_socket_client('tcp://127.0.0.1:5678', $errCode, $errMsg, 10);
    // 推送的数据，包含uid字段，表示是给这个uid推送
    $data = ['tag' => $noticeType, 'uid' => $uid, 'content' => $content];
    // 发送数据，注意5678端口是Text协议的端口，Text协议需要在数据末尾加上换行符
    fwrite($client, json_encode($data) . "\n");
    // 读取推送结果
    echo fread($client, 8192);
    //断开服务
    fclose($client);

    ob_clean();
  }
}
